package zookeeper

import (
	"fmt"
	"github.com/samuel/go-zookeeper/zk"
	"logagent/utils"
	"time"
)

type ZookeeperMgr struct {
	Conn *zk.Conn
}

type ZkData struct {
	Data    []byte
	Version int32
}

var connected = make(chan struct{})

func callback(evt zk.Event) {
	switch evt.State {
	case zk.StateConnecting:
		fmt.Println("stateConnecting")
		if evt.Err != nil {
			fmt.Println("error:", evt.Err)
		}
		fmt.Println(evt.Server)
	case zk.StateDisconnected:
		fmt.Println("stateDisconnected")
	case zk.StateConnected:
		fmt.Println("stateConnected")
		connected <- struct{}{}
	}
}

func NewZookeeper(hosts []string) *ZookeeperMgr {
	eventCallback := zk.WithEventCallback(callback)
	conn, _, _ := zk.Connect(hosts, 10*time.Second, eventCallback)
	<-connected
	return &ZookeeperMgr{Conn: conn}
}

func (mgr *ZookeeperMgr) Close() {
	mgr.Conn.Close()
}

func (mgr *ZookeeperMgr) Create(path string, data []byte) (string, error) {
	return mgr.Conn.Create(path, data, 0, zk.WorldACL(zk.PermAll))
}

func (mgr *ZookeeperMgr) Get(path string) ([]byte, *zk.Stat, error) {
	return mgr.Conn.Get(path)
}

func (mgr *ZookeeperMgr) Listen(path string) chan ZkData {
	snapshots := make(chan ZkData)

	bytes, stat, events, e := mgr.Conn.GetW(path)
	go func() {
		for {
			if e != nil {
				utils.Printf("监听zookeeper失败: %v", e)
				time.Sleep(time.Second * 10)
			} else {
				snapshots <- ZkData{Data: bytes, Version: stat.Version}
				evt := <-events
				if evt.Path == path && evt.Type == zk.EventNodeDataChanged && evt.Err == nil {
					fmt.Println(evt.Type, evt.Path)
				}
			}
			bytes, stat, events, e = mgr.Conn.GetW(path)
		}
	}()
	return snapshots

}
